package com.xiaohu.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}


/**
 * spark sql处理数据的步骤
 * 1、读取数据源
 * 2、将读取到的DF注册成一个临时视图
 * 3、使用sparkSession的sql函数，编写sql语句操作临时视图，返回的依旧是一个DataFrame
 * 4、将结果写出到hdfs上
 */
object Demo1WordCount {
  def main(args: Array[String]): Unit = {
    //创建编写spark sql的环境
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("sql语法风格编写WordCount")
      .getOrCreate()

    /**
     * spark sql是spark core的上层api，如果要想使用rdd的编程
     * 可以直接通过sparkSession获取SparkContext对象
     */
    //    val context: SparkContext = sparkSession.sparkContext
    //spark sql的核心数据类型是DataFrame
    val df1: DataFrame = sparkSession.read
      .format("csv") // 读取csv格式的文件，但是实际上这种做法可以读取任意分隔符的文本文件
      .option("sep", "\n") //指定读取数据的列与列之间的分隔符
      .schema("line STRING") // 指定表的列字段 包括列名和列数据类型
      .load("spark/data/wcs/words.txt")

    //    println(df1)
    //查看DataFrame的数据内容
    //    df1.show()
    //查看表结构
    //    df1.printSchema()

    /**
     * sql语句是无法直接作用在DataFrame上面的
     * 需要提前将要使用sql分析的DataFrame注册成一张表（临时视图）
     */
    //老版本的做法将df注册成一张表
    //    df1.registerTempTable("wcs")
    df1.createOrReplaceTempView("wcs")

    /**
     * 编写sql语句作用在表上
     * sql语法是完全兼容hive语法
     */
    val df2: DataFrame = sparkSession.sql(
      """
        |select
        |t1.word,
        |count(1) as counts
        |from(
        |select
        |explode(split(line,'\\|')) as word
        |from wcs) t1 group by t1.word
        |""".stripMargin)
    //    df2.show()

    //通过观察源码发现，DataFrame底层数据类型其实就是封装了DataSet的数据类型
    val resDS: Dataset[Row] = df2.repartition(1)

    /**
     * 将计算后的DataFrame保存到本地磁盘文件中
     */
    resDS.write
      .format("csv") //默认的分隔符是英文逗号
      //      .option("sep","\t")
      .mode(SaveMode.Overwrite) // 如果想每次覆盖之前的执行结果的话，可以在写文件的同时指定写入模式，使用模式枚举类
      .save("spark/data/sqlout1") // 保存的路径其实是指定的一个文件夹


  }
}
